//  Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.

// liliupeng@inspur.com

#pragma once

#include "abstract_version_node.h"
#include "db/dbformat.h"
#include "pure_mem/mvcc_key.h"
#include "rocksdb/types.h"

namespace rocksdb {

struct UncodedVersionNode : public VersionNode {
private:
  char *buf_;

private:
  Slice key_;
  Slice value_;
  Slice internal_key_;
  SequenceNumber seq_;
  ValueType type_;
  MvccKey mvccKey_;
  size_t len_;

private:
  void updateMvccKey() {
    assert(!internal_key_.empty());

    mvccKey_.seqNumAndType_ = DecodeFixed64(internal_key_.data()
                                            + internal_key_.size() - 8);
    assert(mvccKey_.seqNumAndType_ == PackSequenceAndType(seq_, type_));

    if (internal_key_.size() == 8) {
      mvccKey_.userKey_ = Slice(internal_key_.data(), 0);
      mvccKey_.timestamp_ = Slice(internal_key_.data(), 0);
      return;
    }

    const char ts_size = key_[key_.size() - 1];
    if (static_cast<size_t>(ts_size) >= key_.size()) {
      mvccKey_.userKey_ = key_;
      mvccKey_.timestamp_ = Slice(key_.data(), 0);
      return;
    }
    mvccKey_.userKey_ = Slice(key_.data(), key_.size() - ts_size - 1);
    mvccKey_.timestamp_ = Slice(mvccKey_.userKey_.data() + mvccKey_.userKey_.size(), ts_size);
    mvccKey_.userKeyWithTime_ = Slice(key_.data(), key_.size() - ts_size);
  }

public:
  UncodedVersionNode(const Slice &key, const Slice &value,
                     SequenceNumber seq, ValueType type)
      : VersionNode(VersionNode::NodeType::UNCODED),
        buf_(nullptr), seq_(seq), type_(type)
  {
    size_t key_size = key.size();
    size_t val_size = value.size();
    size_t internal_key_size = key_size + ENCODED_SEQUENCE_TYPE_LEN;
    len_ = VarintLength(internal_key_size) + internal_key_size
           + VarintLength(val_size) + val_size;
    buf_ = new char[len_];

    char *p = EncodeVarint32(buf_, internal_key_size);
    memcpy(p, key.data(), key_size);
    key_ = Slice(p, key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(seq, type);
    EncodeFixed64(p, packed);
    internal_key_ = Slice(p - key_size, key_size + ENCODED_SEQUENCE_TYPE_LEN);
    p += ENCODED_SEQUENCE_TYPE_LEN;
    p = EncodeVarint32(p, val_size);
    memcpy(p, value.data(), val_size);
    value_ = Slice(p, val_size);

    updateMvccKey();
  }

  ~UncodedVersionNode() {
    delete[] buf_;
  }

  UncodedVersionNode(const UncodedVersionNode &) = delete;
  UncodedVersionNode & operator=(const UncodedVersionNode &) = delete;
  UncodedVersionNode(UncodedVersionNode &&) = delete;
  UncodedVersionNode & operator=(UncodedVersionNode &&) = delete;

public:
  const Slice& GetKey() const {
    return key_;
  }

  const Slice& GetValue() const {
    return value_;
  }

  const MvccKey& GetMvccKey() const {
    return mvccKey_;
  }

  const Slice& GetInternalKey() const {
    return internal_key_;
  }

  SequenceNumber GetSequenceNumber() const {
    return seq_;
  }

  ValueType GetValueType() const {
    return type_;
  }

  size_t GetLen() const {
    return len_;
  }


public:
  const char* Key() const override {
    return buf_;
  }

  void GetInternalKey(Slice& key, const char* buf) override {
    key = internal_key_;
  }

  void GetMvccKey(MvccKey& mk, Slice& key) override {
    mk =  mvccKey_;
  }

  void GetValue(const Slice& key, Slice& value) override {
    value = value_;
  }

  void GetSeqAndType(uint64_t seqNumAndType, uint64_t* seq, ValueType* t) override {
    *seq = seq_;
    *t = type_;
  }
};

}  // namespace rocksdb

